package com.ztest.multithreaddemo1;

import com.apop.MainApplication;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 多线程调度作业Demo
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = MainApplication.class)
@Slf4j
public class MultiThreadTaskDemo {

    @Resource
    private AsyncHttpClientService asyncHttpClientService;
    @Resource
    private DataBaseService dataBaseService;

    @Resource
    private TaskStatusStore taskStatusStore;

    /**
     * 线程池配置
     * @return
     */
    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("async-task-");
        executor.initialize();
        return executor;
    }

    /**
     * 多线程发请求demo
     */
    @Test
    public void sendRequest() throws InterruptedException {

        log.info("生成调度列表=====");
        List<Integer> batchList = Arrays.asList(1,2);
        log.info("开始执行作业=====");

        // 生成任务ID
        int taskId = taskStatusStore.generateTaskId();
        // 设置任务状态
        taskStatusStore.setTaskStatus(taskId, "PENDING");
        // 异步执行任务
        CompletableFuture.runAsync(() -> {
            try {
                List<CompletableFuture<Void>> futures = batchList.stream()
                        .map(batchId -> asyncHttpClientService.sendRequestAsync(batchId)
                                .thenAccept(dataBaseService::saveData)) // 当请求完成时，将结果保存到数据库
                        .collect(Collectors.toList());

                // 等待所有请求完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
                taskStatusStore.setTaskStatus(taskId, "COMPLETED");
            } catch (Exception e) {
                taskStatusStore.setTaskStatus(taskId, "FAILED");
                log.error("任务执行失败", e);
            }
        });

        // 启动一个后台线程来定期轮询任务状态
        startTaskPollerThread(taskId);

        // 返回任务ID给前端
        log.info("任务ID: {}", taskId);
        // 这里可以返回 taskId 给前端

        // 这里为了测试输出结果，让主线程睡眠一段时间
        Thread.sleep(20000);

    }

    private void startTaskPollerThread(int taskId) {

        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

        Runnable pollTask = () -> {
            String status = taskStatusStore.getTaskStatus(taskId);
            log.info("任务ID: {}, 状态: {}", taskId, status);

            if ("COMPLETED".equals(status) || "FAILED".equals(status)) {
                log.info("任务已完成或失败，停止轮询");
                executor.shutdown();
            }
        };

        // 每5秒轮询一次
        executor.scheduleAtFixedRate(pollTask, 0, 5, TimeUnit.SECONDS);
    }

}
